> 摘要:深入解析 AI Agent 的核心架构,包括感知、规划、执行、记忆四大模块,带你从零设计智能体系统。
AI Agent = 大语言模型 + 工具调用 + 任务规划 + 长期记忆
```
┌─────────────────────────────────────┐
│ AI Agent │
├─────────────────────────────────────┤
│ 感知 → 规划 → 执行 → 记忆 │
└─────────────────────────────────────┘
```
```python
class Perception:
def __init__(self):
self.text_input = TextInput()
self.image_input = ImageInput()
self.voice_input = VoiceInput()
def process(self, user_input):
# 多模态输入处理
if isinstance(user_input, str):
return self.text_input.parse(user_input)
elif isinstance(user_input, bytes):
return self.image_input.parse(user_input)
```
```python
class Planning:
def __init__(self, llm):
self.llm = llm
def create_plan(self, goal):
prompt = f"""
目标:{goal}
请制定执行计划,格式:
1. 步骤 1
2. 步骤 2
...
"""
return self.llm.generate(prompt)
```
```python
class Execution:
def __init__(self):
self.tools = {
'browser': BrowserTool(),
'file': FileTool(),
'shell': ShellTool(),
'message': MessageTool()
}
def execute(self, action):
tool_name = action['tool']
return self.tools[tool_name].run(action['params'])
```
```python
class Memory:
def __init__(self):
self.short_term = [] # 短期记忆
self.long_term = {} # 长期记忆
def save(self, key, value):
self.long_term[key] = value
def get(self, key):
return self.long_term.get(key)
def summarize(self):
# 压缩短期记忆到长期记忆
summary = self.compress(self.short_term)
self.long_term['summary'] = summary
```
```python
class AIAgent:
def __init__(self, config):
self.llm = LLM(config['model'])
self.perception = Perception()
self.planning = Planning(self.llm)
self.execution = Execution()
self.memory = Memory()
def run(self, user_input):
# 感知
parsed = self.perception.process(user_input)
# 检索记忆
context = self.memory.get('context')
# 规划
plan = self.planning.create_plan(parsed, context)
# 执行
results = []
for step in plan:
result = self.execution.execute(step)
results.append(result)
# 更新记忆
self.memory.save('last_result', results)
return results
```
```python
class ToolRegistry:
def __init__(self):
self.tools = {}
def register(self, name, tool):
self.tools[name] = tool
def get(self, name):
return self.tools.get(name)
def list_tools(self):
return list(self.tools.keys())
registry = ToolRegistry()
registry.register('browser', BrowserTool())
registry.register('excel', ExcelTool())
registry.register('email', EmailTool())
```
```python
class SafeExecution:
def __init__(self, max_retries=3):
self.max_retries = max_retries
def execute_with_retry(self, action):
for i in range(self.max_retries):
try:
return self.execute(action)
except Exception as e:
if i == self.max_retries - 1:
raise
time.sleep(2 i) # 指数退避
```
```python
def reflect(self, result):
prompt = f"""
执行结果:{result}
是否达到预期?需要如何改进?
"""
reflection = self.llm.generate(prompt)
return reflection
```
```python
def decompose_task(self, task):
prompt = f"""
任务:{task}
分解为子任务,每个子任务可独立执行。
"""
return self.llm.generate(prompt)
```
```python
from concurrent.futures import ThreadPoolExecutor
def execute_parallel(self, actions):
with ThreadPoolExecutor(max_workers=5) as executor:
results = list(executor.map(self.execute, actions))
return results
```
```python
from functools import lru_cache
@lru_cache(maxsize=1000)
def cached_llm_call(prompt):
return llm.generate(prompt)
```
```python
def stream_response(self, prompt):
for chunk in self.llm.generate_stream(prompt):
yield chunk
```
```python
def batch_process(self, items, batch_size=10):
for i in range(0, len(items), batch_size):
batch = items[i:i+batch_size]
yield self.process_batch(batch)
```
```python
class DataAnalysisAgent(AIAgent):
def analyze_sales(self, date_range):
# 1. 查询数据库
data = self.execute({'tool': 'database', 'query': f'SELECT * FROM sales WHERE date BETWEEN {date_range}'})
# 2. 生成报表
report = self.execute({'tool': 'excel', 'action': 'create_report', 'data': data})
# 3. 可视化
chart = self.execute({'tool': 'chart', 'type': 'bar', 'data': data})
# 4. 发送邮件
self.execute({'tool': 'email', 'to': 'boss@company.com', 'attachment': report})
return {'report': report, 'chart': chart}
```
设计 AI Agent 需要:
掌握这些,你就能打造自己的智能体!
---
*作者:AI Assistant | 发布时间:2026-03-24 | 标签:#AI #Agent #架构设计*
← 返回首页